﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using AzureMRCore.DataTypes;
using Microsoft.WindowsAzure.StorageClient;

namespace AzureMRCore.OutputFormat
{
    public class DoubleVectorOutputFormat : IOutputFormat<IntKey, DoubleVectorValue>
    {
        #region IOutputFormat<IntKey,DoubleVectorValue> Members

        public long GetSize(List<KeyValuePair<IntKey, DoubleVectorValue>> pairs)
        {
            if (pairs.Count != 0)
            {
                DoubleVectorValue value = pairs.First().Value;
                int vectorLength = value.Value.Length;
                return (pairs.Count*(vectorLength*8 + 4)) + 9;
            }
            return 0;
        }

        public byte[] ToBytes(List<KeyValuePair<IntKey, DoubleVectorValue>> pairs)
        {
            byte[] bytes = new byte[GetSize(pairs)];
            int length = pairs.First().Value.Value.Length;
            Buffer.BlockCopy(BitConverter.GetBytes(pairs.Count), 0, bytes, 0, 4);
            Buffer.BlockCopy(BitConverter.GetBytes(length), 0, bytes, 4, 4);
            Buffer.BlockCopy(BitConverter.GetBytes(true), 0, bytes, 8, 1);
            int i = 9;
            foreach (KeyValuePair<IntKey, DoubleVectorValue> pair in pairs)
            {
                Buffer.BlockCopy(BitConverter.GetBytes(pair.Key.Key), 0, bytes, i, 4);
                i += 4;
                Buffer.BlockCopy(pair.Value.Value, 0, bytes, i, length*8);
                i += length*8;
            }
            return bytes;
        }

        public void FromBytes(byte[] values, ConcurrentDictionary<IntKey, List<DoubleVectorValue>> decodedData)
        {
            int count = BitConverter.ToInt32(values, 0);
            int length = BitConverter.ToInt32(values, 4);
            bool keyed = BitConverter.ToBoolean(values, 8);

            for (int i = 0; i < count; i++)
            {
                int keyValue = BitConverter.ToInt32(values, (i*((length*8) + 4)) + 9);
                IntKey intKey = IntKey.GetInstance(keyValue);

                double[] vector = new double[length];
                Buffer.BlockCopy(values, (i*((length*8) + 4)) + 13, vector, 0, length*8);

                List<DoubleVectorValue> valueList = decodedData.GetOrAdd(intKey, new List<DoubleVectorValue>());

                DoubleVectorValue doubleVectorValue = DoubleVectorValue.GetInstance(vector);
                lock (valueList)
                {
                    valueList.Add(doubleVectorValue);
                }
            }
        }

        public Dictionary<string, IntKey> UploadValues(List<KeyValuePair<IntKey, DoubleVectorValue>> pairs,
                                                       object firstID, string iteration, CloudBlobContainer container)
        {
            Dictionary<string, IntKey> references = new Dictionary<string, IntKey>();
            StringBuilder buffer = new StringBuilder();

            string fileName = "Out_" + firstID + "_" + iteration + ".bin";
            MemoryStream stream = new MemoryStream();
            BinaryWriter writer = new BinaryWriter(stream);

            int vectorLength = 0;
            if (pairs.Count != 0)
            {
                DoubleVectorValue value = pairs.First().Value;
                vectorLength = value.Value.Length;
            }

            Int32 count = pairs.Count;
            writer.Write(count);
            writer.Write(vectorLength);
            writer.Write(0);
            writer.Write(0);
            writer.Write(true);

            foreach (KeyValuePair<IntKey, DoubleVectorValue> pair in pairs)
            {
                writer.Write(pair.Key.Key);
                for (int i = 0; i < vectorLength; i++)
                {
                    writer.Write(pair.Value.Value[i]);
                }
            }

            CloudBlockBlob resultBlob = container.GetBlockBlobReference(fileName);
            container.GetPageBlobReference(fileName);
            writer.Flush();
            stream.Seek(0, SeekOrigin.Begin);
            resultBlob.UploadFromStream(stream);
            references.Add(resultBlob.Uri.AbsoluteUri, default(IntKey));
            return references;
        }

        public void DecodeData(string fileName, string key,
                               ConcurrentDictionary<IntKey, List<DoubleVectorValue>> decodedData)
        {
            BinaryReader reader = new BinaryReader(File.Open(fileName, FileMode.Open));
            int numVectors = reader.ReadInt32();
            int vectorLength = reader.ReadInt32();
            int row = reader.ReadInt32();
            int rowOffset = reader.ReadInt32();
            bool isKeyed = reader.ReadBoolean();
            long counter = 0;

            while (counter < numVectors)
            {
                int keyValue = reader.ReadInt32();
                IntKey intKey = IntKey.GetInstance(keyValue);

                double[] vector = new double[vectorLength];
                for (var i = 0; i < vectorLength; i++)
                {
                    vector[i] = reader.ReadDouble();
                }

                List<DoubleVectorValue> values = decodedData.GetOrAdd(intKey, new List<DoubleVectorValue>());

                DoubleVectorValue doubleVectorValue = DoubleVectorValue.GetInstance(vector);
                lock (values)
                {
                    values.Add(doubleVectorValue);
                }

                counter++;
            }
        }

        #endregion
    }
}